package com.intct.dwd;

import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author gufg
 * @since 2025-07-02 15:38
 */
public class DwdSQLApplication {
    public static void main(String[] args) throws Exception {
        // 配置属性
        Configuration conf = new Configuration();
        conf.set(RestOptions.BIND_PORT, "8081");

        // 创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(1);
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        tEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60 + 5));

        tEnv.executeSql("CREATE TABLE area_info (" +
                " `op` string," +
                " `after` MAP<string, string>," +
                " `table` string," +
                "  `ts_ms` bigint," +
                "  pt as proctime()," +
                "  et as to_timestamp_ltz(ts_ms, 0), " +
                "  watermark for et as et - interval '3' second " +
                ") WITH (" +
                " 'connector' = 'kafka'," +
                " 'topic' = 'ods_databases'," +
                " 'properties.bootstrap.servers' = 'cdh-node:9092'," +
                " 'properties.group.id' = 'ok_goup_id'," +
                " 'scan.startup.mode' = 'earliest-offset'," +
                " 'format' = 'json'" +
                ")");

        Table table = tEnv.sqlQuery("select `after`['code'] as code," +
                "`after`['name'] as name" +
                " from area_info where `table` = 'area_info'");

        table.execute().print();

    }
}
